package com.sunyur.tailor.channel.service.middle.kafka;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @description kafka消息监听
 */
@Component
@ConditionalOnProperty(name = "spring.kafka.switch", havingValue = "open")
public class KafkaConsumer {

    private static final Logger LOGGER_INFO = LoggerFactory.getLogger(KafkaConsumer.class);
    private static final Logger LOGGER_ERROR = LoggerFactory.getLogger(KafkaConsumer.class);

    /**
     * mdm-kafka监控消息
     */
    @KafkaListener(topics = "${spring.kafka.mdm.consumer.topic}", containerFactory = "${spring.kafka.mdm.consumer.container-factory}")
    public void mdmConsume(String message, Acknowledgment ack) {
        LOGGER_INFO.info("kafka [mdm] param:{}",JSON.toJSONString(message));
        //调用业务类，消费消息

        //执行成功增加偏移量
        ack.acknowledge();
    }
}
